/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package edu.csl.study.spark.source.pool

import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.{BlockingQueue, LinkedBlockingDeque}

import org.apache.spark.internal.Logging

import scala.util.control.NonFatal
// TODO_MA 注释：
/**
 * An event loop to receive events from the caller and process all events in the event thread. It
 * will start an exclusive event thread to process all events.
 *
 * Note: The event queue will grow indefinitely. So subclasses should make sure `onReceive` can
 * handle events in time to avoid the potential OOM.
 */
private[spark] abstract class EventLoop[E](name: String) extends Logging {
    
    private val eventQueue: BlockingQueue[E] = new LinkedBlockingDeque[E]()
    
    private val stopped = new AtomicBoolean(false)
    
    /**
     * TODO_MA 马中华 https://blog.csdn.net/zhongqi2513
     *   注释： driver 中，初始化了饿一个  dagSchedudelr
     *   它里面又初始化了一个 eventThread 专门用来处理  JobSubmitted
     */
    // Exposed for testing.
    private[spark] val eventThread = new Thread(name) {
        setDaemon(true)
        
        override def run(): Unit = {
            try {
                while (!stopped.get) {
                    
                    // TODO_MA 注释：获取消息
                    // TODO_MA 注释：一定要注意：当 sparkContext 还没有初始化好的时候，是不执行 sc.runJob 提交任务的。
                    // TODO_MA 注释：当执行 sc.runJob(sc) 的时候，就会提交 Job 到这儿来。
                    val event = eventQueue.take()
                    try {
                        /**
                         * TODO_MA 马中华 https://blog.csdn.net/zhongqi2513
                         *   注释：根据事件的类型，调用不同的 handleXXX 方法来进行处理。
                         *   当接收到任务提交的时候： event = JobSubmitted
                         */
                        onReceive(event)
                    } catch {
                        case NonFatal(e) => try {
                            onError(e)
                        } catch {
                            case NonFatal(e) => logError("Unexpected error in " + name, e)
                        }
                    }
                }
            } catch {
                case ie: InterruptedException => // exit even if eventQueue is not empty
                case NonFatal(e) => logError("Unexpected error in " + name, e)
            }
        }
        
    }
    
    /**
     * TODO_MA 马中华 https://blog.csdn.net/zhongqi2513
     *   注释：
     */
    def start(): Unit = {
        
        if (stopped.get) {
            throw new IllegalStateException(name + " has already been stopped")
        } // Call onStart before starting the event thread to make sure it happens before onReceive
        /**
         * TODO_MA 马中华 https://blog.csdn.net/zhongqi2513
         *   注释：
         */
        onStart()
        
        // TODO_MA 注释：跳转到到 它的 run() 方法
        eventThread.start()
    }
    
    def stop(): Unit = {
        if (stopped.compareAndSet(false, true)) {
            eventThread.interrupt()
            var onStopCalled = false
            try {
                eventThread.join() // Call onStop after the event thread exits to make sure onReceive happens before onStop
                onStopCalled = true
                onStop()
            } catch {
                case ie: InterruptedException => Thread.currentThread().interrupt()
                    if (!onStopCalled) {
                        // ie is thrown from `eventThread.join()`. Otherwise, we should not call `onStop` since
                        // it's already called.
                        onStop()
                    }
            }
        } else {
            // Keep quiet to allow calling `stop` multiple times.
        }
    }
    
    /**
     * Put the event into the event queue. The event thread will process it later.
     */
    def post(event: E): Unit = {
        eventQueue.put(event)
    }
    
    /**
     * Return if the event thread has already been started but not yet stopped.
     */
    def isActive: Boolean = eventThread.isAlive
    
    /**
     * Invoked when `start()` is called but before the event thread starts.
     */
    protected def onStart(): Unit = {}
    
    /**
     * Invoked when `stop()` is called and the event thread exits.
     */
    protected def onStop(): Unit = {}
    
    /**
     * Invoked in the event thread when polling events from the event queue.
     *
     * Note: Should avoid calling blocking actions in `onReceive`, or the event thread will be blocked
     * and cannot process events in time. If you want to call some blocking actions, run them in
     * another thread.
     */
    protected def onReceive(event: E): Unit
    
    /**
     * Invoked if `onReceive` throws any non fatal error. Any non fatal error thrown from `onError`
     * will be ignored.
     */
    protected def onError(e: Throwable): Unit
    
}
